package advertising

import advertising.analyzer.{BaseKpiAnalyzer, UserTagsAnalyzer}
import advertising.handler.{FieldValueHandler, TypeCastHandler}
import advertising.sql.HiveDataSourceSQL
import org.apache.spark.SparkConf
import org.apache.spark.sql.{SparkSession}

//  .--,       .--,
// ( (  \.---./  ) )
//  '.__/o   o\__.'
//     {=  ^  =}
//      >  -  <
//     /       \
//    //       \\
//   //|   .   |\\
//   "'\       /'"_.-~^`'-.
//      \  _  /--'         `
//    ___)( )(___
//   (((__) (__)))    高山仰止, 景行行止; 虽不能至, 心向往之.
//                    High mountains, scenery. Although not to, the heart yearns for it.
object Driver {

  def main(args: Array[String]): Unit = {

    if(args.length < 1){
      System.err.print("未配置数据输入路径")
      System.exit(1)
    }
    // 处理数据时间
    val handingTime = args(0)

    val conf = new SparkConf()
      .setAppName("precision-advertising-system")
    val spark = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()

    import spark.implicits._
    // 从 Hive 中获取 ods 数据
    val adRawData = spark.sql(HiveDataSourceSQL.odsAdRawData(handingTime))
    // 原始数据表
    adRawData.createOrReplaceTempView("ad_raw_data")
    adRawData
      .map(row => TypeCastHandler.row2BaseTabData(row))
      .createOrReplaceTempView("base_tab_data")

    // 地域点击量统计
    BaseKpiAnalyzer.analyzeGeoClickedCountCollectAndSave2Hive(spark)
    // 地域分布统计
    BaseKpiAnalyzer.analyzeGeoTabDataCollectAndSave2Hive(spark)
    // 运营商分布统计
    BaseKpiAnalyzer.analyzeOperatorTabDataCollectAndSave2Hive(spark)
    // 网络类型分布统计
    BaseKpiAnalyzer.analyzeNetworkTypeTabDataCollectAndSave2Hive(spark)
    // 设备类型分布统计
    BaseKpiAnalyzer.analyzeDeviceTypeTabDataCollectAndSave2Hive(spark)
    // 系统类型汇总
    BaseKpiAnalyzer.analyzeOSTypeTabDataCollectAndSave2Hive(spark)
    // 应用名称汇总
    BaseKpiAnalyzer.analyzeAppNameTabDataCollectAndSave2Hive(spark)

    // 构建基础数据 RDD
    val adRawInfoRDD = adRawData.rdd

    val baseAdInfoRDD = adRawInfoRDD.filter(row => {
      FieldValueHandler.getAllDriverId(row).length != 0
    }).map(row => {
      val driversId = FieldValueHandler.getAllDriverId(row)
      (driversId, row)
    }).cache()

    // 用户标签
    UserTagsAnalyzer.collectUserTagsAndSave2HBase(baseAdInfoRDD)

    spark.stop()
  }

}
